package com.binarysoft.cmdhandler.common.mq;

import java.util.Collection;
import java.util.Enumeration;
import java.util.Map;
import java.util.StringTokenizer;

import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import com.binarysoft.service.common.mq.ConnectTask;
import com.binarysoft.service.common.mq.MQDestination;
import com.binarysoft.service.common.mq.MQService;
import com.binarysoft.service.common.mq.MessageQueueService;
import common.framework.command.handler.AbstractCommandHandler;
import common.framework.command.helper.CommandHelper;
import common.framework.command.server.Terminal;
import common.framework.dsb.annotation.DSBCmd;
import common.framework.dsb.client.DefaultServiceFactory;
import common.framework.dsb.client.ServiceFactory;

@DSBCmd("mq")
public class MQCmdHandler extends AbstractCommandHandler {

	@Override
	public void handle(String args, Terminal terminal) throws Exception {
		terminal.println("");
		StringTokenizer stier = new StringTokenizer(args);
		int count = stier.countTokens();
		if (count < 2) {
			showOptions(terminal);
			return;
		}
		stier.nextToken();
		String cmd = stier.nextToken();
		if ("pool".equalsIgnoreCase(cmd)) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			Map<String, MQService> pools = messageQueueService.listPools();

			String[] tableHAD = { "Name", "Value" };
			String[] tableGAP = { "----", "-----" };
			int[] format = { 34, 30 };
			terminal.println(CommandHelper.format(tableHAD, format));
			terminal.println(CommandHelper.format(tableGAP, format));

			for (MQService pool : pools.values()) {

				terminal.println(CommandHelper.format(new Object[] { "Pool name:", pool.getPoolName() }, format));
				terminal.println(CommandHelper.format(new Object[] { "Broker URL:", pool.getBrokerUrl() }, format));
				terminal.println(CommandHelper.format(new Object[] { "ExpiryTimeout:", pool.getExpiryTimeout() }, format));
				terminal.println(CommandHelper.format(new Object[] { "IdleTimeout:", pool.getIdleTimeout() }, format));
				terminal.println(CommandHelper.format(new Object[] { "MaxActiveSessionPerConnection:", pool.getMaxActiveSessionPerConnection() }, format));
				terminal.println(CommandHelper.format(new Object[] { "MaxConnections:", pool.getMaxConnections() }, format));
				terminal.println(CommandHelper.format(new Object[] { "IsUseAsynSend:", pool.isUseAsyncSend() }, format));
				terminal.println(CommandHelper.format(new Object[] { "IsCreateConnectionOnStartup:", pool.isCreateConnectionOnStartup() }, format));
				terminal.println(CommandHelper.format(new Object[] { "IsBlockIfSessionPoolIsFull:", pool.isBlockIfSessionPoolIsFull() }, format));
				terminal.println("");
			}

		} else if ("dest".equalsIgnoreCase(cmd)) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			Map<String, MQDestination> destinations = messageQueueService.listDestinations();

			terminal.println("ack_mode:       1 -- AUTO_ACKNOWLEDGE  2 -- CLIENT_ACKNOWLEDGE");
			terminal.println("delivery_mode:  1 -- NON_PERSISTENT    2 -- PERSISTENT");
			terminal.println("");

			String[] tableHAD = { "key", "queue", "ack_mode", "delivery_mode", "is transacted", "is topic" };
			String[] tableGAP = { "---", "-----", "--------", "-------------", "--------------", "--------" };
			int[] format = { 35, 32, 10, 14, 18, 16 };
			terminal.println(CommandHelper.format(tableHAD, format));
			terminal.println(CommandHelper.format(tableGAP, format));

			for (Map.Entry<String, MQDestination> entry : destinations.entrySet()) {
				String key = entry.getKey();
				MQDestination dest = entry.getValue();
				Object[] row = { key, dest.getQueueName(), dest.getDeliveryMode(), dest.getDeliveryMode(), dest.isIsTransaction(), dest.isIsTopic() };
				terminal.println(CommandHelper.format(row, format));
			}

		} else if ("conn".equalsIgnoreCase(cmd)) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			Collection<ConnectTask> tasks = messageQueueService.listReconnectTask();
			String[] tableHAD = { "Pool name", "Queue name" };
			String[] tableGAP = { "---------", "----------" };
			int[] format = { 34, 30 };
			terminal.println(CommandHelper.format(tableHAD, format));
			terminal.println(CommandHelper.format(tableGAP, format));

			for (ConnectTask task : tasks) {
				Object[] row = { task.getPoolName(), task.getQueueName() };
				terminal.println(CommandHelper.format(row, format));
			}

		} else if ("pending".equalsIgnoreCase(cmd)) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			int pendingSize = messageQueueService.pendingMsgs();
			terminal.println("Number of pending messages: " + pendingSize);

		} else if ("send".equalsIgnoreCase(cmd) && count == 5) {
			String poolName = stier.nextToken();
			String queue = stier.nextToken();
			String text = stier.nextToken();
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			messageQueueService.send(poolName, queue, text);
		} else if ("recv".equalsIgnoreCase(cmd) && count == 4) {
			String poolName = stier.nextToken();
			String queueName = stier.nextToken();
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");

			messageQueueService.receive(poolName, queueName, new DefaultMessageListener(terminal));
		} else if ("stop".equalsIgnoreCase(cmd) && count == 4) {
			String poolName = stier.nextToken();
			String queueName = stier.nextToken();
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			MessageQueueService messageQueueService = serviceFactory.lookup("DSB/MessageQueueService");
			messageQueueService.stop(poolName, queueName);
		} else {
			showOptions(terminal);
		}
		terminal.println("");
	}

	private void showOptions(Terminal terminal) throws Exception {
		terminal.println(" Usage:");
		terminal.println(" mq pool                                      -- list all MQ pools.");
		terminal.println(" mq dest                                      -- list all MQ destinations.");
		terminal.println(" mq conn                                      -- list all MQ consumer reconnect tasks.");
		terminal.println(" mq pending                                   -- show number of pending messages.");
		terminal.println(" mq send   <pool_name> <queue_name>  <text>   -- send text message.");
		terminal.println(" mq recv   <pool_name> <queue_name>           -- create listener to receive message.");
		terminal.println(" mq stop   <pool_name> <queue_name>           -- stop to receive message.");
		terminal.println("");
	}

	private class DefaultMessageListener implements MessageListener {

		private Terminal terminal = null;

		public DefaultMessageListener(Terminal terminal) {
			super();
			this.terminal = terminal;
		}

		@Override
		public void onMessage(Message message) {
			try {
				if (message instanceof TextMessage) {
					TextMessage tMsg = (TextMessage) message;
					System.out.println("Text: " + tMsg.getText());
				} else if (message instanceof MapMessage) {

					MapMessage mapMsg = (MapMessage) message;

					Enumeration<String> enums = mapMsg.getMapNames();

					while (enums.hasMoreElements()) {
						String keyName = enums.nextElement();
						Object value = mapMsg.getObject(keyName);
						System.out.println(keyName + " = " + value);
					}

				} else {
					System.out.println("recv: " + message.toString());
				}
			} catch (Exception e) {
				e.printStackTrace(System.out);
			}
		}// onMessage
	}
}
